package com.yuntsg.ruiijn.paperana.test;

import cn.hutool.core.collection.ListUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import com.yuntsg.ruiijn.paperana.utils.SmallTool;
import com.yuntsg.ruiijn.paperana.utils.esutils.EsServerEntityTempNew;
import com.yuntsg.ruiijn.paperana.utils.esutils.IndexConfig;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
@SuppressWarnings("all")
public class ElasticsearchSaveDataText {


    // es 入库代码

    public static void main(String[] args) throws IOException, SQLException {


//        String TABLE_NAME_PRE = "ncbi_pdf_to_txt_";
//        if (args[1].equals("wos")) {
//            TABLE_NAME_PRE = "wos_pdf_to_txt_";
//        }
//
//        System.out.println(TABLE_NAME_PRE);
//        // 创建低级客户端
//        RestClient restClient = RestClient
//                .builder(new HttpHost("10.1.1.11", 9200))
//                .setRequestConfigCallback(re -> {
//                    re.setConnectTimeout(10000);
//                    re.setSocketTimeout(150000);
//                    re.setConnectionRequestTimeout(20000);
//                    return re;
//                })
//                .setHttpClientConfigCallback(re -> {
//                    re.setMaxConnTotal(1000);
//                    re.setMaxConnPerRoute(1000);
//                    return re;
//                })
//                .build();
//
//        // 使用Jackson映射器创建传输层
//        ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
//        // 创建API客户端
//        ElasticsearchClient client = new ElasticsearchClient(transport);
//
//        Integer dbType = TABLE_NAME_PRE.startsWith("ncbi") ? 1 : 2;
//
//        createIndex(client);

//        DataSource ds4 = DSFactory.get("ds3");
//        Db db = Db.use(ds4);

//        DruidDataSource ds2 = new DruidDataSource();
//        //jdbc:mysql://119.188.182.156:3306
//        ds2.setUrl("jdbc:mysql://10.1.1.15:11306/splitdate?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai");
//        ds2.setUsername("root");
//        ds2.setPassword("quanfang3.1415");
//        Db db = Db.use(DSFactory.get("youhua"));
//
//        List<String> tables = db.query("show tables", String.class);
//
//        String finalTABLE_NAME_PRE = TABLE_NAME_PRE;
//
//        tables.stream().filter(table -> table.startsWith(finalTABLE_NAME_PRE)).forEach(TABLE_NAME -> {
//            long count = 0;
//            try {
//                count = db.count(Entity.create(TABLE_NAME));
//            } catch (SQLException e) {
//                throw new RuntimeException(e);
//            }
//
//            if (count > 0) {
//
////        count = 5000;
//
//                int size = 500;
//                int page = PageUtil.totalPage((int) count, size);
//
//                List<PageInfo> pageInfos = new ArrayList<>();
//                for (Integer i = 0; i < page; i++) {
//                    Integer from = i * size + 1;
//                    Integer to = (i + 1) * size;
//                    pageInfos.add(new PageInfo(from, to));
//                }
//                System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");
//
//
//                pageInfos.parallelStream().forEach(p -> {
//                    TimeInterval timerInner = DateUtil.timer();
//
//                    List<Entity> query = null;
//                    try {
//                        query = db.query("SELECT * FROM " + TABLE_NAME + " WHERE  id BETWEEN ? AND ?", p.getFrom(), p.getTo());
//                    } catch (SQLException e) {
//                        throw new RuntimeException(e);
//                    }
//                    if (null != query && query.size() > 0) {
//                        List<BulkOperation> list = new ArrayList<>();
//                        int num = 0;
//                        for (Entity entity : query) {
//                            if (entity.getStr("str").length() < 512) {
//                                Map<String, Object> doc = new HashMap<>();
//                                doc.put("paragraph_name", entity.getStr("paragraph_name"));
//                                doc.put("sub_project", entity.getStr("sub_project"));
//                                doc.put("sentence", entity.getInt("sentence"));
//                                doc.put("file_id", entity.getInt("file_id"));
//                                doc.put("database_type", dbType);
//                                doc.put("table_id", entity.getInt("table_id"));
//                                doc.put("str", entity.getStr("str"));
//                                list.add(new BulkOperation.Builder().create(builder -> builder.id(IdUtil.fastSimpleUUID()).index(INDEX_NAME).document(doc)).build());
//                                num++;
//                            }
//
//                        }
//                        if (num<400){
//                            log.info("{}___{}，数据读取完成 ,数量{}", DateUtil.now(), p.toString(), list.size());
//                        }
//
//                        log.info("{}___{}，数据读取完成 ,数量{}", DateUtil.now(), p.toString(), list.size());
//
//                        try {
//                            BulkResponse response = client.bulk(builder -> builder.index(INDEX_NAME).operations(list));
//                            log.info(String.valueOf(response.errors()));
//                        } catch (Exception e) {
//                            log.error("插入失败:{}", e.getMessage());
//                        }
//                    }
//                });
//                log.info("数据写入完成:{}", DateUtil.now());
//            }
//        });
//
//        List<String> strs = new ArrayList<>();
//
//        Optional<String> any = strs.parallelStream().filter(s -> {
//            return s.equals("2");
//        }).findFirst();
//
//        if (any.isPresent()) {
//            String s = any.get();
//        }

    }

    /**
     * @param pdf 解析出来的待比对文章段落
     * @param rid 记录表的主键id
     * @param uid 用户id
     * @return
     */
    public static boolean dealManualDataStaNew(List<EsInsertNew> strs, ElasticsearchClient client, String esIndex) {
        String index = esIndex;
//        log.info("入库es名称+" + index);
        List<String> strsCompare = new ArrayList<>();

        // 删除存在的数据
        //进行不对然后出结果
//        EsClient.deleEsDataIndex(index);
        boolean isTrue = true;
        List<BulkOperation> bulkOperations = new ArrayList<>();
        for (EsInsertNew str : strs) {
            EsServerEntityTempNew esEntity = new EsServerEntityTempNew();
            // 临时新增一个年份 查询快速 不再使用多一层查询
            // String  paragraph_name
            // String sub_project
            // String str
            // Integer code;
            // Integer year;
            // Integer database_type;
            //  Integer file_id;
            //  Integer table_id;
            // String pmid_ut;
            // Integer paragraph;
            // Integer sentence;
            esEntity.setParagraph_name(str.getParagraph_name());
            esEntity.setSub_project(str.getSub_project());
            esEntity.setStr(str.getStr());
            esEntity.setCode(str.getCode());
            esEntity.setYear(str.getYear());
            esEntity.setDatabase_type(str.getDatabase_type());
            esEntity.setFile_id(str.getFile_id());
            esEntity.setTable_id(str.getTable_id());
            esEntity.setPmid_ut(str.getPmid_ut());
            esEntity.setParagraph(str.getParagraph());
            esEntity.setSentence(str.getSentence());
            bulkOperations.add(new BulkOperation.Builder().create(d -> d.document(esEntity)).build());
        }
        if (bulkOperations.size() > 5000) {
            List<List<BulkOperation>> split = ListUtil.split(bulkOperations, 5000);
            for (List<BulkOperation> operations : split) {
                try {
                    client.bulk(e -> e.index(index).operations(operations));
                } catch (IOException e) {
                    isTrue = false;
                    e.printStackTrace();
                }
            }
        } else {
            try {
                client.bulk(e -> e.index(index).operations(bulkOperations));
            } catch (IOException e) {
                isTrue = false;
                e.printStackTrace();
            }
        }
        return isTrue;
    }


    // 2024年7月29日15:49:45 新版es 测试添加入库
    @SneakyThrows
    public static void createIndexNew(ElasticsearchClient client, String esIndex) {
        if (client.indices().exists(b -> b.index(esIndex)).value()) {
            log.info("es 存在不进行创建++" + esIndex);
            return;
        } else {
            //  2个副本 6个分片
            CreateIndexResponse response = client.indices().create(builder -> builder
                    .settings(indexSettingsBuilder -> indexSettingsBuilder.numberOfReplicas("1").numberOfShards("12"))
                    .mappings(typeMappingBuilder -> typeMappingBuilder
                            .properties("paragraph_name", i -> i.text(p -> p))
                            .properties("sub_project", i -> i.text(p -> p))
                            .properties("str", t -> t.text(p -> p
                                    .analyzer("ik_max_word")
                                    .searchAnalyzer("ik_max_word")
                                    .fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))
                                    .fields("suggest", l -> l.completion(u -> u.analyzer("english")))
                                    .fields("wildcard", v -> v.wildcard(a -> a))

                            ))
                            .properties("code", b -> b.integer(p -> p))
                            .properties("year", b -> b.integer(p -> p))
                            .properties("database_type", i -> i.integer(p -> p))
                            .properties("file_id", b -> b.integer(p -> p))
                            .properties("table_id", b -> b.integer(p -> p))
                            // 新增了一个字段
                            .properties("pmid_ut", b -> b.text(p -> p.fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))))
                            .properties("paragraph", b -> b.integer(p -> p))
                            .properties("sentence", i -> i.integer(p -> p))
                    )
                    .index(esIndex));
            SmallTool.printMess("es不存在 创建完毕");
            log.info("acknowledged={}", response.acknowledged());
        }

    }

    // 2024年7月29日15:49:45 新版es 测试添加入库
    @SneakyThrows
    public static void createIndexNew2(ElasticsearchClient client, String esIndex) {
        if (client.indices().exists(b -> b.index(esIndex)).value()) {
            log.info("es 存在不进行创建++" + esIndex);
            return;
        }
        CreateIndexResponse response = client.indices().create(builder -> builder
                .settings(indexSettingsBuilder -> indexSettingsBuilder.numberOfReplicas("0").numberOfShards("32"))
                .mappings(typeMappingBuilder -> typeMappingBuilder
                        .properties("paragraph_name", b -> b.text(p -> p.fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))))
                        .properties("sub_project", b -> b.text(p -> p.fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))))
                        .properties("sentence", b -> b.integer(p -> p))
                        .properties("file_id", b -> b.integer(p -> p))
                        // 新增了一个字段
                        .properties("pmid_ut", b -> b.text(p -> p.fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))))
                        .properties("database_type", b -> b.integer(p -> p))
                        .properties("table_id", b -> b.integer(p -> p))
                        .properties("str", t -> t.text(p -> p
                                .analyzer("ik_max_word")
                                .searchAnalyzer("ik_max_word")
                                .fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))
                                .fields("suggest", l -> l.completion(u -> u.analyzer("english")))
                                .fields("wildcard", v -> v.wildcard(a -> a))

                        ))
                )
                .index(esIndex));
        log.info("acknowledged={}", response.acknowledged());
    }


    public static void createIndex(ElasticsearchClient client) throws IOException {

        if (client.indices().exists(b -> b.index(IndexConfig.PUB_TEXT_INDEX)).value()) {
            return;
        }
        CreateIndexResponse response = client.indices().create(builder -> builder
                .settings(indexSettingsBuilder -> indexSettingsBuilder.numberOfReplicas("0").numberOfShards("32"))
                .mappings(typeMappingBuilder -> typeMappingBuilder
                        .properties("paragraph_name", b -> b.text(p -> p.fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))))
                        .properties("sub_project", b -> b.text(p -> p.fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))))
                        .properties("sentence", b -> b.integer(p -> p))
                        .properties("file_id", b -> b.integer(p -> p))
                        .properties("database_type", b -> b.integer(p -> p))
                        .properties("table_id", b -> b.integer(p -> p))
                        .properties("str", t -> t.text(p -> p
                                .analyzer("ik_max_word")
                                .searchAnalyzer("ik_max_word")
                                .fields("keyword", r -> r.keyword(m -> m.normalizer("lowercase")))
                                .fields("suggest", l -> l.completion(u -> u.analyzer("english")))
                                .fields("wildcard", v -> v.wildcard(a -> a))

                        ))
                )
                .index(IndexConfig.PUB_TEXT_INDEX));
        log.info("acknowledged={}", response.acknowledged());
    }

}
